package com.dtwave.cfstreaming.process;

import com.alibaba.fastjson.JSONObject;
import com.dtwave.cfstreaming.bean.ProcessData;
import com.dtwave.cfstreaming.process.function.MyProcessFunction;
import com.dtwave.cfstreaming.process.function.ProcessDataFilter;
import com.dtwave.cfstreaming.process.function.ProcessMapFunction;
import com.dtwave.cfstreaming.process.selector.KeyByFunction;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static org.apache.flink.table.api.Expressions.$;

public class TableRegister {

    public static void streamTableFromKafka(StreamTableEnvironment tableEnv, SingleOutputStreamOperator<JSONObject> flatMapStream) {
        SingleOutputStreamOperator<ProcessData> process = flatMapStream
                //从kafka数据中过滤出需要创建实时表的数据
                .filter(new ProcessDataFilter())
                //让数据依据某个字段进行分组，类似于groupby是延时器的前置条件
                .keyBy(new KeyByFunction())
                //延时器，对满足条件的数据进行延迟发送
                .process(new MyProcessFunction())
                .flatMap(new ProcessMapFunction());

        //将流数据处理为表并注册
        Table processTable = tableEnv.fromDataStream(process,
                $("name"),
                $("age"),
                $("password"),
                $("sex"),
                $("source"),
                /**
                 *事件时间，如果要两张表进行关联join计算，需要考虑时态的统一性，实时表传过来的某个时间点去获取对应时间的维度的数据，
                 * 如果不加处理时间字段，程序运行时会直接将维表所有的数据拉到程序中进行状态的保存，并且不会进行实时的维表更新，
                 * 拿到的数据就一致都是滞后的
                 */
                $("proctime").proctime()
        );
        tableEnv.createTemporaryView("processTab",processTable);
    }
    public static void dimTableFromHbase(StreamTableEnvironment tableEnv, ParameterTool params) {
        //创建维度表
        tableEnv.executeSql("建表语句，flinkSql用于创建关联其他数据源的维度表，方便在flinksql中进行关联计算，具体创建方式查看官网");
    }
}
